package task

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sync/atomic"
	"time"
)

// taskScheduled 定时任务队列调度器
type taskScheduled struct {
	taskManager *TaskManager
	taskQueue   *taskQueue    // 任务队列信息
	taskMaps    *taskMaps     // 任务基本信息
	timer       *time.Timer   // 任务队列消费触发定时器
	nextTimeMs  *atomic.Int64 // 队列中下一个任务执行时间的毫秒时间戳
	isRuning    *atomic.Bool
}

// timerSignal 触发定时器运行
func (s *taskScheduled) timerSignal(task *Task) {
	if !s.taskManager.isRuning.Load() || task == nil {
		return
	}
	// 任务执行时间等于或小于当前时间
	if !task.executeAt.After(time.Now()) {
		if s.timer != nil {
			s.nextTimeMs.Store(0)
			s.timer.Stop()
		}
		s.runQueue()
		return
	}
	// 获取任务执行时间毫秒时间戳, 不存储时间对象，避免重复创建时间对象
	executeMs := task.executeAt.UnixMilli()
	if s.nextTimeMs.Load() == 0 || executeMs < s.nextTimeMs.Load() {
		s.nextTimeMs.Store(executeMs)
		if s.timer == nil {
			s.timer = time.AfterFunc(time.Until(task.executeAt), func() {
				s.nextTimeMs.Store(0)
				s.runQueue()
			})
		} else {
			s.timer.Stop()
			s.timer.Reset(time.Until(task.executeAt))
		}
	}
}

// runQueue 执行任务队列
func (s *taskScheduled) runQueue() {
	if !s.isRuning.CompareAndSwap(false, true) {
		// 正在运行中，无需处理信号
		return
	}
	defer s.isRuning.Store(false)
	timeNow := time.Now()
	task, next := s.taskQueue.PopAndPeekNext(timeNow)
	for task != nil {
		if !s.taskManager.isRuning.Load() {
			// Scheduled 未运行(或已关闭)，无需处理信号
			return
		}
		// 执行任务
		s.taskManager.workerCh <- task
		// 继续处理下一个任务
		if next == nil {
			return
		}
		if next.executeAt.After(timeNow) {
			// next.executeAt > 当前时间
			// 没有到期任务, 重置定时器
			s.timerSignal(next)
			return
		}
		task, next = s.taskQueue.PopAndPeekNext(timeNow)
	}
}
func (s *taskScheduled) runWorker() {
	s.taskManager.wg.Add(1)
	go func() {
		defer s.taskManager.wg.Done()
		for task := range s.taskManager.workerCh {
			s.executeTask(task)
		}
	}()
}

// executeTask 执行单个任务
func (s *taskScheduled) executeTask(task *Task) {
	// 检查任务执行状态
	if task == nil || !s.taskManager.isRuning.Load() {
		return
	}
	if !atomic.CompareAndSwapInt32(&task.state, TaskStatePending, TaskStateRunning) {
		// 任务状态不是 TaskStatePending, 说明任务正在执行中或已执行完成或已取消
		return
	}
	// 启动1个任务执行协程
	s.taskManager.wg.Add(1)
	go func() {
		defer s.taskManager.wg.Done()
		ctx, cancel := context.WithTimeout(context.Background(), task.timeout)
		defer cancel()
		done := make(chan struct{}, 1)
		go func() {
			defer func() {
				defer close(done)
				// Panic 异常捕获
				if err := recover(); err != nil {
					// 更新任务状态
					s.updateTaskStatus(task, TaskStateFailed, fmt.Errorf("panic: %v", err))
				}
				// queueRun 触发 executeTask 是通过Pop方式消费的任务信息
				// 这里不需要执行从队列中移除任务的工作
				// shard.taskQueue.Remove(task.key)
			}()
			// 更新周期性任务信息, 重新插入到队列中, 等待下一次执行
			if !task.isOnce {
				s.taskManager.updateIntervalTask(task)
			}
			// 执行任务回调
			if err := task.handler(); err != nil {
				s.updateTaskStatus(task, TaskStateFailed, err)
			} else {
				s.updateTaskStatus(task, TaskStateFinished, nil)
				s.taskManager.log(task.pc,
					slog.LevelDebug,
					"task finished",
					slog.String("key", task.key))
			}
		}()
		select {
		case <-done:
			atomic.CompareAndSwapInt32(&task.state, TaskStateRunning, TaskStatePending)
		case <-ctx.Done():
			atomic.CompareAndSwapInt32(&task.state, TaskStateRunning, TaskStatePending)
			s.updateTaskStatus(task, TaskStateTimeout, errors.New("timeout: "+task.timeout.String()))
		}
	}()
}

func (s *taskScheduled) updateTaskStatus(task *Task, state TaskState, err error) {
	if err != nil {
		s.taskManager.log(task.pc, slog.LevelError, "task error",
			slog.String("key", task.key),
			slog.String("error", err.Error()))
	}
	task.scheduled.taskMaps.Update(task.key, func(taskInfo *TaskInfo) {
		taskInfo.RunCount++
		taskInfo.LastState = state
		taskInfo.LastRunAt = task.executeAt
		taskInfo.LastConsuming = time.Since(task.executeAt)
		taskInfo.LastError = err
		switch state {
		case TaskStateFinished:
			s.taskManager.monitor.IncFinish(taskInfo)
		case TaskStateTimeout:
			s.taskManager.monitor.IncTimeout(taskInfo)
		case TaskStateFailed:
			s.taskManager.monitor.IncFail(taskInfo)
		}
	})
}
